package com.imooc.jiangzh.kafka.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.internals.Topic;

import java.util.*;


public class AdminSample {
    public final static String TOPIC_NAME="jz_topic";

    //psvm
    public static void main(String[] args) throws Exception {
        AdminClient adminClient=AdminSample.adminClient();
        System.out.println("adminClient:"+adminClient);
        //创建topic实例
        createTopic();
        //获取topic列表
        topicLists();
        //删除topic实例
        delTopic();
        //描述topic
        describeTopic();
        //修改Config
        alterConfig();
        //查看Config
        describeConfig();
        //增加partition数量
        incrPartition(2);
    }


    /*
         增加partition数量
     */
    public static void incrPartition(int partitions) throws Exception {
        AdminClient adminClient=adminClient();
        Map<String,NewPartitions> partitionsMap=new HashMap<>();
        NewPartitions newPartitions=NewPartitions.increaseTo(partitions);
        partitionsMap.put(TOPIC_NAME,newPartitions);
        CreatePartitionsResult clientPartitions = adminClient.createPartitions(partitionsMap);
        clientPartitions.all().get();


    }


    /*
         修改config信息
     */
    public static void alterConfig() throws Exception {
        AdminClient adminClient=adminClient();
//        Map<ConfigResource,Config> configMaps=new HashMap<>();
//        //组织两个参数
//        ConfigResource configResource =new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
//        Config config=new Config(Arrays.asList(new ConfigEntry("preallocate","true")));
//        configMaps.put(configResource,config);
//        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configMaps);
/*
        从2.3以上的版本新修改的API
     */
        Map<ConfigResource,Collection<AlterConfigOp>> configMaps=new HashMap<>();
//        //组织两个参数
        ConfigResource configResource =new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        AlterConfigOp alterConfigOp = new AlterConfigOp(new ConfigEntry("preallocate","false"),AlterConfigOp.OpType.SET);
        configMaps.put(configResource,Arrays.asList(alterConfigOp));
        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configMaps);
        alterConfigsResult.all().get();

    }

    public static void describeConfig() throws Exception {
        AdminClient adminClient=adminClient();
        ConfigResource configResource= new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        // TODO 这里做预留，集群时会讲到
//        ConfigResource configResource= new ConfigResource(ConfigResource.Type.BROKER,TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult= adminClient.describeConfigs(Arrays.asList(configResource));
        Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
        configResourceConfigMap.entrySet().stream().forEach((entry)->{
            System.out.println("configResource:"+entry.getKey()+", Config");
        });

    }



    /*
         描述topic
     */
    public static void describeTopic() throws Exception {
        AdminClient adminClient=adminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        entries.stream().forEach(entry ->{
            System.out.println("name:"+entry.getKey()+", desc:"+entry.getValue());
        } );
    }

    /*
         删除topic
     */
    public static void delTopic() throws Exception {
        AdminClient adminClient=adminClient();
        DeleteTopicsResult deleteTopicsResult= adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        deleteTopicsResult.all().get();
    }


    /*
         获取topic列表
     */
    public static void topicLists() throws Exception {
        AdminClient adminClient=adminClient();
        //是否查看internal选项
        ListTopicsOptions options=new ListTopicsOptions();
        options.listInternal(true);
        ListTopicsResult listTopicsResult=adminClient.listTopics(options);
        //ListTopicsResult listTopicsResult=adminClient.listTopics();
        Set<String> names= listTopicsResult.names().get();
        Collection<TopicListing> topicListings =listTopicsResult.listings().get();
        KafkaFuture<Map<String,TopicListing>> mapKafkaFuture =listTopicsResult.namesToListings();
        //打印names
        names.stream().forEach(System.out::println);
        topicListings.stream().forEach((topicList)->{
            System.out.println(topicList);
        });

    }




    /*
         创建topics实例
     */
    public static void createTopic() {
        AdminClient adminClient=adminClient();
        //副本因子
        Short rs=1;
        NewTopic newTopic=new NewTopic(TOPIC_NAME,1,rs);
        CreateTopicsResult topics= adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println("CreateTopicsResult:"+topics);
    }
    /*
     设置AdminClient
     */
    public static AdminClient adminClient(){
        Properties properties=new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
        AdminClient adminClient =AdminClient.create(properties);
        return adminClient;

    }
}
